package com.biz.primus.ms.member.service;

import com.biz.primus.base.vo.PageableRequestVo;
import com.biz.primus.common.utils.DateUtil;
import com.biz.primus.model.member.inter.INotice;
import com.biz.primus.model.member.vo.NoticeVo;
import com.biz.primus.model.member.vo.request.BasePageableRequestVo;
import com.biz.primus.model.member.vo.request.NoticePageRequestVo;
import com.biz.primus.model.member.vo.response.BaseResponseVo;
import com.biz.primus.model.push.enums.NotifyType;
import com.biz.primus.model.push.vo.NoticePlatform;
import com.biz.primus.model.push.vo.Notification;
import com.biz.primus.model.push.vo.PushMessage;
import com.biz.primus.ms.base.service.AbstractBaseService;
import com.biz.primus.ms.member.dao.po.Notice;
import com.biz.primus.ms.member.dao.redis.NoticeRedisDao;
import com.biz.primus.ms.member.dao.redis.NotificationRedisDao;
import com.biz.primus.ms.member.dao.repository.NoticeRepository;
import com.biz.primus.ms.member.dao.ro.MemberRo;
import com.biz.primus.ms.member.dao.ro.NoticeRo;
import com.biz.primus.ms.member.dao.ro.NotificationRo;
import com.biz.primus.ms.member.trans.NoticeRoToNoticeVo;
import com.biz.primus.ms.member.trans.NoticeToNoticeRo;
import com.biz.primus.ms.member.trans.NoticeToNoticeVo;
import com.biz.primus.ms.member.trans.NoticeVoToNotice;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.*;

/**
 * 消息推送服务
 * @author jun.liu
 * @date 2017年10月23日
 * @reviewer
 * @see
 */
@Slf4j
@Service
public class NoticeService extends AbstractBaseService
{

    private final NoticeRedisDao noticeRedisDao;

    private final NoticeRepository noticeRepository;

    private final NotificationRedisDao notificationRedisDao;

    private final MemberService memberService;

    @Autowired
    public NoticeService(NoticeRedisDao noticeRedisDao, NoticeRepository noticeRepository,
                                 NotificationRedisDao notificationRedisDao, MemberService memberService)
    {
        this.noticeRedisDao = noticeRedisDao;
        this.noticeRepository = noticeRepository;
        this.notificationRedisDao = notificationRedisDao;
        this.memberService = memberService;
    }

    /**
     * 给某个用户发通知
     * @author jun.liu
     * @param vo
     * @param userId
     */
    @Transactional(rollbackFor = Exception.class)
    public void sendToUser(NoticeVo vo, Long userId, String channelCode)
    {
        MemberRo user = memberService.findRoById(userId);
        Notice po = save(vo);
        noticeRedisDao.sendToUser(po.getId(), po.getCreateTimestamp(), userId);
        PushMessage pushMessage = getBasePushMessage(vo);
        sendNotification(userId, pushMessage);
    }

    @NotNull
    private Notice save(NoticeVo vo)
    {
        Notice po = new NoticeVoToNotice().apply(vo);
        po.setId(idService.getNextId());
        saveOrUpdateUsingPo(noticeRepository, noticeRedisDao, po, new NoticeToNoticeRo());
        return po;
    }

    /**
     * 构建发送消息体
     * @param notice
     * @return
     */
    private PushMessage getBasePushMessage(INotice notice)
    {
        PushMessage pushMessage = new PushMessage();
        pushMessage.setAlertMsg(notice.getContent());
        pushMessage.setUrl(notice.getUrl());
        return pushMessage;
    }

    private String getNoticePlatform(String userLastAgent)
    {
        if (StringUtils.equalsIgnoreCase(userLastAgent, "appstore")) {
            return NoticePlatform.IOS;
        } else if (StringUtils.containsIgnoreCase(userLastAgent, "mi")) {
            return NoticePlatform.MI;
        } else {
            return NoticePlatform.ANDROID;
        }
    }

    private void sendNotification(Long userId, PushMessage pushMessage)
    {
        Notification notification = new Notification();
        NotificationRo notificationRo = notificationRedisDao.findByUserId(userId);
        if (log.isDebugEnabled())
        {
            log.debug("find notificationRo {} by  userId: {}", notificationRo, userId);
        }
        notification.setPlatform(getNoticePlatform(notificationRo.getUserLastAgent()));
        notification.setNotifyType(NotifyType.P2P);
        notification.setPushMessage(pushMessage);
        notification.setTarget(notificationRo.getLastLoginDeviceToken());
        //todo 此处调用发消息 下列注释的是以前的代码片段
//        BizMessag<?> message = SimpleBizMessage.newMessage(notification);
//        messageService.sendMessage(BizBaseQueue.PUSH_QUEUE, message);
    }

    /**
     * 给一群人发通知
     * @author jun.liu
     * @param vo
     * @param userIds
     */
    @Transactional(rollbackFor = Exception.class)
    public void sendToUsers(NoticeVo vo, List<Long> userIds)
    {
        Notice po = save(vo);
        if (log.isDebugEnabled())
        {
            log.debug("send notice {}  to users: {}", po.getId(), userIds);
        }
        noticeRedisDao.sendToUsers(po.getId(), po.getCreateTimestamp(), userIds);
        PushMessage pushMessage = getBasePushMessage(vo);
        for (Long userId : userIds)
        {
            sendNotification(userId, pushMessage);
        }
    }

    /**
     * 给所有人发通知
     * @author jun.liu
     * @param vo
     */
    @Transactional(rollbackFor = Exception.class)
    public void sendToEveryOne(NoticeVo vo)
    {
        Notice po = save(vo);
        if (log.isDebugEnabled()) {
            log.debug("send notice {}  to everyone", po.getId());
        }
        noticeRedisDao.sendToEveryOne(po.getId());
        PushMessage pushMessage = getBasePushMessage(vo);
        publishNotification(pushMessage);
    }

    private void publishNotification(PushMessage pushMessage)
    {
        Notification notification = new Notification();
        notification.setNotifyType(NotifyType.BROADCAST);
        notification.setPlatform(NoticePlatform.ANDROID_IOS);
        notification.setPushMessage(pushMessage);
        //todo 此处调用发消息 下列注释的是以前的代码片段
//        BizMessage<?> message = SimpleBizMessage.newMessage(notification);
//        messageService.sendMessage(BizBaseQueue.PUSH_QUEUE, message);
    }

    /**
     * 查询发出的通知
     * @author jun.liu
     * @param vo
     * @return
     */
    public List<NoticeVo> findNormal(BasePageableRequestVo vo)
    {
        if (log.isDebugEnabled())
        {
            log.debug("find notice po {}", vo);
        }
        Pageable pageable;
        //传入参数有误时，默认返回第一页数据
        if(vo == null)
        {
            pageable = new PageRequest(0,10);
        }
        else
        {
            pageable = new PageRequest(vo.getPage(),vo.getSize());
        }
        List<Notice> list = noticeRepository.findByOrderByCreateTimestampDesc(pageable);
        List<NoticeVo> resVoList = Lists.newArrayList();
        if(CollectionUtils.isNotEmpty(list)){
            resVoList = Lists.transform(list,new NoticeToNoticeVo());
        }
        return resVoList;
    }

    /**
     * 查询发出的通知用于后台
     * @author devil
     * @param reqVo
     * @return
     */
//    BootstrapTablePageResul<NoticeVo> findAllNotice(PageVo reqVo);

    /**
     * 获取一个通知详情
     * @author jun.liu
     * @param id
     * @return
     */
    public NoticeVo findOne(Long id)
    {
        Notice notice = noticeRepository.findOne(id);
        NoticeVo resVo = new NoticeVo();
        if(notice != null)
        {
            resVo.setUrl(notice.getUrl());
            resVo.setTitle(notice.getTitle());
            resVo.setContent(notice.getContent());
        }
        return resVo;
    }

    /**
     * 删除对应的通知
     */
    @Transactional(rollbackFor = Exception.class)
    public void delete(Long id)
    {
        notificationRedisDao.delete(id);
        noticeRedisDao.removeNoticeIdOfAllUsersFromSortedSet(id);
        noticeRedisDao.delete(id);
        noticeRepository.delete(id);
    }

    /**
     * 获取用户未收通知数量
     * @author jun.liu
     * @param userId
     * @return
     */
    public long getUserRemainNoticeCount(Long userId)
    {
        //个人通知
        long count = noticeRedisDao.getUserRemainNoticeCount(userId);
        if (log.isDebugEnabled()) {
            log.debug("user {}  self notice count {}", userId, count);
        }
        //公告通知
        Long ts = getUserLastReceiveTs(userId);
        count += noticeRedisDao.getEveryOneRemainNoticeCount(ts);
        return count;
    }

    /**
     * 返回用户最后收到通知时间，如果为空 默认返回一个月前（此处的规则只影响公告通知）
     */
    private Long getUserLastReceiveTs(Long userId) {
        Long ts = noticeRedisDao.getUserLastReceiveTs(userId);
        if (ts == null) {
            ts = System.currentTimeMillis() - DateUtil.DAY30;
            if (log.isDebugEnabled()) {
                log.debug("user {}  last recivie ts is not set set it :{}", userId, ts);
            }
        }
        if (log.isDebugEnabled()) {
            log.debug("user:{}  last recivie ts {} notice po {}", userId, ts);
        }
        return ts;
    }

    /**
     * 获取用户未收通知
     * @author jun.liu
     * @param userId
     * @param delNotice TODO
     * @return
     */
    public List<NoticeVo> getUserRemainNotices(Long userId, boolean delNotice) {
        List<NoticeRo> result = new ArrayList<>();
        //个人通知
        List<NoticeRo> my = noticeRedisDao.getUserRemain(userId);
        if (log.isDebugEnabled()) {
            log.debug("user {}  self notice size {}", userId, my == null ? 0 : my.size());
        }
        if (CollectionUtils.isNotEmpty(my)) {
            result.addAll(my);
        }

        //公告通知
        Long ts = getUserLastReceiveTs(userId);
        List<NoticeRo> all = noticeRedisDao.getEveryOneRemainNotices(ts);
        if (log.isDebugEnabled()) {
            log.debug("user {}  every one notice size {}", userId, all == null ? 0 : all.size());
        }
        if (CollectionUtils.isNotEmpty(all)) {
            result.addAll(all);
        }
        Collections.sort(result);
        if(delNotice){
            //将用户自己的通知删除
            noticeRedisDao.deleteUserNotice(userId);
            //保存已经接受的公告消息时间--当前时间
            noticeRedisDao.setUserLastReceiveTs(userId, System.currentTimeMillis());
        }
        List<NoticeVo> resVoList = Lists.newArrayList();
        if(CollectionUtils.isNotEmpty(result)){
            resVoList = Lists.transform(result,new NoticeRoToNoticeVo());
        }
        return resVoList;
    }

    /**
     * 获取用户未收的 [推送给所有人] 的通知
     * @param userId
     * @param size
     */
    public List<NoticeVo> getUserRemainNotices(Long userId, Integer size){
        List<NoticeRo> result = new ArrayList<>();
        //该用户最后拉取 消息列表的时间
        Long ts = getUserLastReceiveTs(userId);
        List<NoticeRo> all = noticeRedisDao.getEveryOneRemainNotices(ts);
        if (CollectionUtils.isNotEmpty(all)) {
            Collections.sort(all, new Comparator<NoticeRo>() {
                @Override
                public int compare(NoticeRo o1, NoticeRo o2) {
                    return -o1.getCreateTimestamp().compareTo(o2.getCreateTimestamp());
                }
            });
            for(int i = 0; i < size && i < all.size() ; i++) {
                result.add(i, all.get(i));
            }
        }
        List<NoticeVo> resVoList = Lists.newArrayList();
        if(CollectionUtils.isNotEmpty(result)){
            resVoList = Lists.transform(result,new NoticeRoToNoticeVo());
        }
        return resVoList;
    }

    /**
     * 分页获取消息(暂只获取向所有人推送的消息)
     * @param vo
     * @return
     */
    public List<NoticeVo> getAllNoticeRo(NoticePageRequestVo vo){
        Long userId = vo.getUserId();
        if(userId != null) {
            //删除推送给个人的消息
            noticeRedisDao.deleteUserNotice(userId);
            //公共未读消息是获取   (该用户最后拉取消息列表的时间,+inf)内的消息
            //设置 该用户最后拉取消息列表的时间
            noticeRedisDao.setUserLastReceiveTs(userId, System.currentTimeMillis());
        }
        List<NoticeRo> result = noticeRedisDao.getAllNotice(vo.getPage(),vo.getSize());
        List<NoticeVo> resVoList = Lists.newArrayList();
        if(CollectionUtils.isNotEmpty(result)){
            resVoList = Lists.transform(result,new NoticeRoToNoticeVo());
        }
        return resVoList;
    }

    public static String noticestr = "新版上线，活动期内快喝订单首单立减10元！";

    public List<String> getNoLoginNoticemsg(){
        //初始化消息 此处暂时写死 以后可能会从后台配置获取
        List<String> noticemsg = new ArrayList<>();
        noticemsg.add(noticestr);
        return noticemsg;
    }

    public List<String> getLoginNoticemsg(Long userId){
        List<String> noticemsg = new ArrayList<>();
        List<NoticeVo> roList = this.getUserRemainNotices(userId, false);
        if(CollectionUtils.isEmpty(roList))
        {
            noticemsg.add(noticestr);
        }
        else
        {
            //暂时只取一个
            NoticeVo vo = roList.get(0);
            noticemsg.add(vo.getTitle());
        }
        return noticemsg;
    }

    /**
     * 把所有用于最后读消息的时间设置为当前时间,达到"清除"所有未读消息的目的
     */
    public void cleanRemaindNotice()
    {
        log.info("查询用户全部数量...");
        long total = memberService.findMemberCount();
        log.info("查询所有用户完毕,开始分组清除消息...");
        //启动10个线程跑
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("cleanRemaindNotice-pool-%d").build();
        ExecutorService executorService = new ThreadPoolExecutor(10, 200,
                0L, TimeUnit.MICROSECONDS,
                new LinkedBlockingQueue<Runnable>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
        int size = 1500;
        int totalPages = (int) Math.ceil(total/Double.valueOf(size));
        for(int page = 0; page < totalPages ; page++){
            final BasePageableRequestVo pageable = new BasePageableRequestVo(page,size);
            executorService.execute(new Runnable() {
                @Override
                public void run(){
                    try{
                        List<BaseResponseVo> lists = memberService.findBaseInfoByPageable(pageable);
                        if(CollectionUtils.isEmpty(lists)){
                            log.info("The list data is null...");
                            return;
                        }
                        Long userId = null;
                        for(BaseResponseVo vo : lists) {
                            userId = vo.getMemberId();
                            if (userId != null) {
                                noticeRedisDao.deleteUserNotice(userId);
                                noticeRedisDao.setUserLastReceiveTs(userId, System.currentTimeMillis());
                            }
                        }
                    }catch (Exception e){

                    }
                }
            });
        }
        executorService.shutdown();
    }

}
